package com.demo.kafka.flink.test;
  
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;  
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/*
* 打开终端，执行nc -l -p 9999，在 idea 运行以上代码，并在终端中依次发送下面的信息：

king 35.6
wang 36.3
zhao 36.7
lee 35.5
liu 37.0
* */
public class FilterTaskExample {
    public static void main(String[] args) throws Exception {  
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();  
  
        // 注意：在Java中，socketTextStream 通常用于 Streaming API，但这里为了示例，我们假设这是一个DataStream示例  
        // 实际应用中，你可能需要改为使用 DataStreamSource 或其他适合的DataStream创建方式  
        DataStreamSource<String> dataStream = env.socketTextStream("localhost", 9999);  
  
        DataStream<People> peopleStream = dataStream  
                .map(new MapFunction<String, People>() {  
                    @Override  
                    public People map(String line) throws Exception {  
                        String[] arr = line.split(" ");  
                        return new People(arr[0], Float.parseFloat(arr[1]));  
                    }  
                });  
  
        // Flink Java API 并不直接支持类似 Scala 的 split 和 select 方法。  
        // 一种替代方案是使用 filter 操作符来实现类似的功能。  
        DataStream<People> feverStream = peopleStream  
                .filter(people -> people.temperature > 36.0f);  
  
        DataStream<People> normalStream = peopleStream  
                .filter(people -> people.temperature <= 36.0f);

        feverStream.print();

        env.execute("Select Operator in Java...");  
    }  
  
    public static class People {  
        public String name;  
        public float temperature;  
  
        public People(String name, float temperature) {  
            this.name = name;  
            this.temperature = temperature;  
        }  
  
        @Override  
        public String toString() {  
            return "People{" +  
                    "name='" + name + '\'' +  
                    ", temperature=" + temperature +  
                    '}';  
        }  
    }  
}